//////////////////////////////////////////////////////////////////////////////////////////////////////
// 项目名   : ThreadPool
// 文件名   : ThreadPool.h
// 作者     : Aiye
// 日期     : 2025/9/28 07:10
// 邮箱     : 2878548868@qq.com
// 摘要     : 
//////////////////////////////////////////////////////////////////////////////////////////////////////
#ifndef THREADPOOL_THREADPOOL_H
#define THREADPOOL_THREADPOOL_H
#include <algorithm>
#include <condition_variable>
#include <deque>
#include <functional>
#include <future>
#include <optional>
#include <thread>
#include <vector>

/**
 * @brief 线程池类用于管理和调度线程执行任务
 */
class ThreadPool
{
public:
    /**
     * @brief 构造函数，初始化线程数量
     * @param nThreads 初始线程数，默认为硬件支持的最大并发数
     * @param nCapacity 任务列表的最大容量，
     */
    explicit ThreadPool( std::size_t nThreads = std::thread::hardware_concurrency() ? std::thread::hardware_concurrency() : 4
        , std::size_t nCapacity = 0)
            : _nCapacity(nCapacity)
    {
        _vecWorkers.reserve(nThreads); // 预分配容器
        for (std::size_t i = 0 ; i < nThreads ; ++i)
        {
            _vecWorkers.emplace_back([this](std::stop_token st)
            {
                WorkerLoop(st); // 初始化容器
            });
        }
    }

    /**
     * @brief 删除默认复制构造函数，防止对象被复制
     */
    ThreadPool(const ThreadPool& other) = delete;

    /**
     * @brief 删除默认赋值运算符，防止对象被赋值
     */
    ThreadPool& operator=(const ThreadPool& other) = delete;

    ~ThreadPool()
    {
        RequestStop(); // 请求所有工作线程停止
    }

public:
    /**
     * @brief 提交一个任务到线程池，并返回一个 std::future 对象，如果任务队列已满该函数会阻塞，直到有空位置可用。
     * @tparam F 可调用对象类型
     * @tparam Args 可调用对象参数类型
     * @param f 可调用对象
     * @param args 可调用对象参数
     * @return std::future 对象，用于获取任务的执行结果
     */
    template<class F , class... Args>
    auto Submit(F&& f , Args&&... args) -> std::future<std::invoke_result_t<F , Args...>>
    {
        using R = std::invoke_result_t<F , Args...>; // 获取可调用对象的返回类型
        std::packaged_task<R()> task{std::bind(std::forward<F> (f), std::forward<Args>(args)...)};
        std::future<R> fut = task.get_future(); // 获取任务的结果

        {
            std::unique_lock lock{_mx}; // 锁定
            if (_nStopping)
            {
                throw std::runtime_error("ThreadPool is stopping"); // 检查是否停止线程池
            }

            if (_nCapacity > 0)
            {
                _cvNotFull.wait(lock , [this]()
                {
                    return _nStopping || _deTasks.size() < _nCapacity;
                });

                if (_nStopping)
                {
                    throw std::runtime_error("ThreadPool is stopping"); // 再次检查是否停止线程池
                }
            }

            _deTasks.emplace_back([pt = std::move(task)]() mutable
            {
                pt(); // 将任务添加到队列中去
            });
        }
        _cv.notify_one(); // 通知有新的任务加入
        return fut; // 返回future对象
    }

    /**
     * @brief 尝试在短时间内将一个任务提交到线程池中
     * @tparam Rep 超过时间的表示类型
     * @tparam Period 超过时间的周期类型
     * @tparam F 可调用对象类型
     * @tparam Args 可调用对象参数的类型
     * @param d 超过时间
     * @param f 可调用对象
     * @param args 传递给可调用对象的参数
     * @return 提交任务成功返回包含 std::optional的std::future ,失败则返回std::nullopt
     */
    template<class Rep , class Period , class F , class... Args>
    auto TrySubmitFor(const std::chrono::duration<Rep , Period>& d , F&& f , Args&&... args)
        -> std::optional<std::future<std::invoke_result_t<F , Args...>>>
    {
        using R = std::invoke_result_t<F , Args...>;
        std::packaged_task<R()> task{std::bind(std::forward<F>(f) , std::forward<Args>(args)...)};
        std::future<R> fut = task.get_future(); // 获取任务的结果

        {
            std::unique_lock lock{_mx}; // 锁定

            if (_nStopping)
            {
                return std::nullopt; // 返回不存在
            }

            if (_nCapacity > 0 && _deTasks.size() >= _nCapacity)
            { // 尝试提交
                if (!_cvNotFull.wait_for(lock , d , [this]{return _nStopping || _deTasks.size() < _nCapacity;}))
                {
                    return std::nullopt;
                }

                if (_nStopping)
                {
                    return std::nullopt; // 再次确认
                }
            }
            _deTasks.emplace_back([pt = std::move(task)]() mutable
            {
               pt(); // 提交任务到队列中
            });
        }
        _cv.notify_one();
        return std::make_optional(std::move(fut));
    }

    /**
     * @brief 等待队列空且无活动任务
     */
    void WaitIdle()
    {
        std::unique_lock lock{_mx};
        _cvIdle.wait(lock ,[this]
        {
           return _deTasks.empty() && _nActive == 0;
        });
    }

    void ShutdownGracefully()
    {
        RequestStop();
    }

private:
    /**
     * @brief 请求线程池停止
     */
    void RequestStop()
    {
        {
            std::lock_guard lock{_mx};
            _nStopping = true;
        }
        _cv.notify_all(); // 通知所有正在等待条件变量，线程池已经停止
        _cvNotFull.notify_all();
    }

    /**
     * @brief 工作线程的主循环
     * @param st 停止令牌
     */
    void WorkerLoop(std::stop_token st)
    {
        while (true)
        {
            std::function<void()> job{}; // 初始化任务对象

            {
                std::unique_lock lock{_mx};
                _cv.wait(lock ,[this , st]()
                {
                    return !_deTasks.empty() || _nStopping || st.stop_requested();
                });

                if (_nStopping || st.stop_requested() && _deTasks.empty())
                {
                    break;
                }

                job = std::move(_deTasks.front()); // 取出任务对象
                _deTasks.pop_front();
                ++_nActive;

                if (_nCapacity > 0 )
                {
                    _cvNotFull.notify_one();
                }
            }

            job(); // 执行任务

            {
                std::lock_guard lock{_mx};
                --_nActive;
                if (_deTasks.empty() && _nActive)
                {
                    _cvIdle.notify_all();
                }
            }
        }
    }

private:
    std::vector<std::jthread> _vecWorkers{};        // 工作线程容器
    std::deque<std::function<void()>> _deTasks{};   // 待执行任务队列

    mutable std::mutex _mx{};                       // 互斥锁
    std::condition_variable _cv{};                  // 任务可用条件变量
    std::condition_variable _cvNotFull{};           // 任务队列未满条件变量
    std::condition_variable _cvIdle{};              // 无活动任务条件变量

    bool _nStopping{false};                         // 停止标志
    std::size_t _nCapacity{};                       // 任务队列最大容量
    std::size_t _nActive{};                         // 当前活跃线程数
};

#endif //THREADPOOL_THREADPOOL_H
